package cn.azzhu.day03

import cn.azzhu.day02.{SensorReading, SensorSource}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 温度平均值，全窗口聚合函数
 *
 * @author azzhu
 * @create 2020-09-21 11:18:42
 */
object AvgTempByProcessWindowFunction {
  case class AvgInfo(id:String,avgTemp:Double,windowStart:Long,windowEnd:Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.addSource(new SensorSource)

    stream.keyBy(_.id)
      .timeWindow(Time.seconds(5))
      .process(new AvgTempFunc)
      .print()

    env.execute("AvgTempByProcessWindowFunction")
  }

  //todo 相比增量聚合函数，缺点是：要保存窗口中的所有元素
  //增量聚合函数只需要保存一个累加器就行了
  //todo  优点：全窗口聚合函数可以访问窗口信息
  class AvgTempFunc extends ProcessWindowFunction[SensorReading,AvgInfo,String,TimeWindow] {
    //在窗口闭合时调用
    override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[AvgInfo]): Unit = {
      val count = elements.size   //窗口闭合时，温度一共有多少条
      var sum = 0.0           //总的温度值
      for (r <- elements) {
        sum += r.temperature
      }
      //单位毫秒
      val windowStart = context.window.getStart
      val windowEnd = context.window.getEnd

      out.collect(AvgInfo(key,sum/count,windowStart,windowEnd))
    }
  }
}
